package com.kele.kettle;


import com.kele.algorithm.AlgorithmUtils;
import com.kele.ddl.CreateTable;
import com.kele.entity.DbModel;
import com.kele.entity.TableDetails;
import com.kele.job.FindDb;
import com.kele.utils.DbDetailsUtils;
import org.junit.Test;
import org.pentaho.di.core.KettleEnvironment;
import org.pentaho.di.core.database.DatabaseMeta;
import org.pentaho.di.core.exception.KettleException;
import org.pentaho.di.repository.kdr.KettleDatabaseRepository;
import org.pentaho.di.repository.kdr.KettleDatabaseRepositoryMeta;
import org.pentaho.di.trans.Trans;
import org.pentaho.di.trans.TransHopMeta;
import org.pentaho.di.trans.TransMeta;
import org.pentaho.di.trans.step.StepMeta;
import org.pentaho.di.trans.step.StepMetaInterface;
import org.pentaho.di.trans.steps.tableinput.TableInputMeta;
import org.pentaho.di.trans.steps.tableoutput.TableOutput;
import org.pentaho.di.trans.steps.tableoutput.TableOutputMeta;

import java.sql.SQLException;
import java.util.List;
import java.util.Scanner;

/**
 * @author lch
 * @version 1.0
 * @date 2020/12/29 9:42
 * @description kettle操作练习
 */
public class KettleDev01 {

    //static DbModel dbModel=new DbModel();
    static TableDetails tableDetails=new TableDetails();
    //过了算法之后的数据
    static String sha1;
    static TableOutput output;


    /**
     *
     * @description
    kettle的初始化
     * @return
     */

    public  static  KettleDatabaseRepository KettleDev01() throws KettleException {
        if (!KettleEnvironment.isInitialized()) {
            try {
                KettleEnvironment.init();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        /*
        数据库元连接
        参数介绍
        kettle资源库名称，数据库类型，连接类型，ip，数据库名称，端口，用户名，密码
         */
        DatabaseMeta databaseMeta = new DatabaseMeta("ETL", "mysql", "Native(JDBC)", "192.168.241.1", "demo?useUnicode=true&characterEncoding=utf-8&useSSL=false",
                "3306", "root", "Aki@lch");

        //设置kettle资源库以数据库形式的资源库元对象
        KettleDatabaseRepositoryMeta repository = new KettleDatabaseRepositoryMeta();
        //保存数据库连接至元对象种
        repository.setConnection(databaseMeta);
        //数据库形式的资源库对象
        KettleDatabaseRepository databaseRepository = new KettleDatabaseRepository();
        //用资源库元对象初始化资源库对象
        databaseRepository.init(repository);
        //连接到资源库
        databaseRepository.connect("admin","admin");
        //判断状态为false | true
        if (databaseRepository.isConnected()) {
            System.out.println("连接成功，资源库名称为：" + databaseRepository.getDatabaseMeta());
            return databaseRepository;
        } else {
            System.out.println("连接失败，资源库名称为：" + databaseRepository.getDatabaseMeta());
            return null;
        }
    }


    /**
     * 创建资源库作业或转换并执行
     * 实例：创建转换 并实现表输入表输出步骤
     */

    public  void demo1() throws KettleException, SQLException {

        //获得资源库对象
        KettleDatabaseRepository repository=KettleDev01();
        //创建转换对象
        TransMeta transMeta=new TransMeta();
        //转换元资源库的名称
        transMeta.setRepository(repository);
        //设置转换的名称
        transMeta.setName("trans01");
        //创建转换的保存的位置
        transMeta.setRepositoryDirectory(repository.findDirectory("/"));
        //添加数据源
        transMeta.addDatabase(repository.getDatabaseMeta());

        //添加表输入组件
        TableInputMeta inputMeta=new TableInputMeta();
        //设置默认值
        inputMeta.setDefault();
        //设置数据源
        inputMeta.setDatabaseMeta(transMeta.findDatabase("ETL"));

        CreateTable.test();

//        String tableName1="user";
//        //调用数据库连接对象
//        tableDetails.setTableName(tableName1);
//        CreateTable.getDbDetails(tableDetails.getTableName());
        //表输出sql语句
        inputMeta.setSQL("select * from "+tableDetails.getTableName());
        //创建步骤实例，把组件放到步骤中
        StepMeta stepMeta1=new StepMeta("步骤1：表输入",inputMeta);
        //把步骤加到作业中
        transMeta.addStep(stepMeta1);



        //添加表输出组件
        TableOutputMeta outputMeta=new TableOutputMeta();
        //设置默认值
        outputMeta.setDefault();
        //设置数据源
        outputMeta.setDatabaseMeta(transMeta.findDatabase("ETL"));
        //输出到那张表
        outputMeta.setTableName("user1");
        //创建步骤实例
        StepMeta stepMeta2=new StepMeta("步骤2：输出到： ",outputMeta);
        //将步骤加到作业里面
        transMeta.addStep(stepMeta2);


        //连接对象，将步骤1和步骤2连接起来,通过hop
        TransHopMeta hopMeta=new TransHopMeta(stepMeta1,stepMeta2);
        //添加到转换中
        transMeta.addTransHop(hopMeta);

        //保存该转换到资源库
        repository.save(transMeta,null);
        //把转换放到执行对象
        Trans trans=new Trans(transMeta);
        //开始执行
        trans.execute(null);
        //等待执行完毕
        trans.waitUntilFinished();
    }


    public static void main(String[] args) throws KettleException, SQLException {
        //获得资源库对象
        KettleDatabaseRepository repository=KettleDev01();
        //创建转换对象
        TransMeta transMeta=new TransMeta();
        //转换元资源库的名称
        transMeta.setRepository(repository);
        //设置转换的名称
        transMeta.setName("trans01");
        //创建转换的保存的位置
        transMeta.setRepositoryDirectory(repository.findDirectory("/"));
        //添加数据源
        transMeta.addDatabase(repository.getDatabaseMeta());

        //添加表输入组件
        TableInputMeta inputMeta=new TableInputMeta();
        //设置默认值
        inputMeta.setDefault();
        //设置数据源
        inputMeta.setDatabaseMeta(transMeta.findDatabase("ETL"));

        String tableName = CreateTable.test();

        //表输入sql语句
        String sql="select * from "+tableName;
        inputMeta.setSQL(sql);
        //创建输入步骤实例，把组件放到步骤中
        StepMeta stepMeta1=new StepMeta("步骤1：表输入",inputMeta);
        //把步骤加到作业中
        transMeta.addStep(stepMeta1);

        //添加表输出组件
        TableOutputMeta outputMeta=new TableOutputMeta();
        //设置默认值
        outputMeta.setDefault();
        //设置数据源
        outputMeta.setDatabaseMeta(transMeta.findDatabase("ETL"));

        List list= DbDetailsUtils.execute(sql);

              for (int i = 0; i <list.size() ; i++) {

            AlgorithmUtils.src= (String) list.get(i);
            sha1 = AlgorithmUtils.jdkSHA1(AlgorithmUtils.src);
        }

        //输出到那张表
        outputMeta.setTableName("t_"+tableName);

        //创建输出步骤实例
        StepMeta stepMeta2=new StepMeta("步骤2：输出到： ",outputMeta);
        //将步骤加到作业里面
        transMeta.addStep(stepMeta2);
        //连接对象，将步骤1和步骤2连接起来,通过hop
        TransHopMeta hopMeta=new TransHopMeta(stepMeta1,stepMeta2);
        //添加到转换中
        transMeta.addTransHop(hopMeta);

        //保存该转换到资源库
        repository.save(transMeta,null);
        //把转换放到执行对象
        Trans trans=new Trans(transMeta);
        //开始执行
        trans.execute(null);
        //等待执行完毕
        trans.waitUntilFinished();
    }

}
